/**
 * Copyright (C) 2012-2013 jumei, Inc.
 * ConnectionManager.java
 * 2015年6月25日
 */
package org.goface.rpc.net;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;

import org.goface.rpc.io.NettyUtils;
import org.goface.rpc.io.RpcClientSideHandler;
import org.goface.rpc.pipeline.ServicePipeLine;

import com.google.common.net.HostAndPort;

/**
 * @author jm110
 */
public class ClientSideConnectionManager {
    private Map<HostAndPort, Channel> channelCache = new ConcurrentHashMap<>();
    private static ClientSideConnectionManager connectionManager = new ClientSideConnectionManager();
    private volatile Bootstrap bootstrap;

    public static ClientSideConnectionManager getInstance() {
        return connectionManager;
    }

    private void initClient(final ServicePipeLine servicePipeLine) {
        if (bootstrap == null) {
            synchronized (this) {
                if (bootstrap == null) {
                    bootstrap = new Bootstrap();
                    bootstrap.group(new NioEventLoopGroup(2, Executors.newFixedThreadPool(2)));
                    bootstrap.channel(NioSocketChannel.class);
                    bootstrap.handler(new ChannelInitializer<Channel>() {
                        @Override
                        protected void initChannel(Channel ch) throws Exception {
                            NettyUtils.addJavaSerialiseCodec(ch, new RpcClientSideHandler(servicePipeLine));
                        }

                    });
                }

            }
        }
    }

    public synchronized Channel createChannel(HostAndPort address, final ServicePipeLine servicePipeLine) {
        initClient(servicePipeLine);
        ChannelFuture future = bootstrap.connect(new InetSocketAddress(address.getHostText(), address.getPort()));
        try {
            future.sync();
        } catch (InterruptedException e) {
            ;
        }
        Channel channel = future.channel();
        channelCache.put(address, channel);
        return channel;
    }

    public Channel getOrCreateServerChannel(final HostAndPort address, final ServicePipeLine servicePipeLine) {
        Channel channel = channelCache.get(address);
        if (channel == null) {
            return createChannel(address, servicePipeLine);
        }
        return channel;
    }

}
